package com.xlh.dokka.svc.core.model;

import com.google.common.util.concurrent.AtomicDouble;
import com.xlh.dokka.svc.core.client.DockerClient;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author cheer
 */
@Data
@Slf4j
public class NodeUsage {

    private Double usedCpus;

    private Long usedMemory;

    private Integer runningCount;

    private static Lock lock = new ReentrantLock();

    public void update(Double cpus, Long memory, Integer count) {
        lock.lock();
        try {
            usedCpus = usedCpus + cpus;
            usedMemory = usedMemory + memory;
            runningCount = runningCount + count;
        } finally {
            lock.unlock();
        }
    }

    public static void sync(List<Node> nodes, ThreadPoolTaskExecutor taskExecutor) {
        CountDownLatch countDownLatch = new CountDownLatch(nodes.size());
        List<Future> futureList = new ArrayList<>(nodes.size());
        for (Node node : nodes) {
            futureList.add(taskExecutor.submit(() -> {
                try {
                    if (node.getHealthy() != null && !node.getHealthy()) {
                        return;
                    }
                    NodeUsage nodeUsage = new NodeUsage();
                    if (nodeUsage.sync(node.getIp(), taskExecutor) && !Thread.interrupted()) {
                        node.setNodeUsage(nodeUsage);
                    }
                } finally {
                    countDownLatch.countDown();
                }
            }));
        }
        try {
            if (!countDownLatch.await(6, TimeUnit.SECONDS)) {
                for (Future future : futureList) {
                    future.cancel(true);
                }
            }
        } catch (InterruptedException e) {
            log.warn("thread " + Thread.currentThread().getName() + " is interrupted");
        }
    }

    public Boolean sync(String ip, ThreadPoolTaskExecutor taskExecutor) {
        List<String> dockerIds = getDockerIds(ip, taskExecutor);
        if (dockerIds != null) {
            return syncNodeUsage(ip, dockerIds, taskExecutor);
        }
        return false;
    }

    private List<String> getDockerIds(String ip, ThreadPoolTaskExecutor taskExecutor) {
        Future<List<String>> future = taskExecutor.submit(() -> DockerClient.listRunningContainers(ip));
        try {
            return future.get(3, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.warn("host {} sync node usage failed", ip, e);
        }
        return null;
    }

    private Boolean syncNodeUsage(String ip, List<String> dockerIds, ThreadPoolTaskExecutor taskExecutor) {
        AtomicInteger runningCount = new AtomicInteger(dockerIds.size());
        AtomicDouble usedCpus = new AtomicDouble(0);
        AtomicLong usedMemory = new AtomicLong(0);
        CountDownLatch countDownLatch = new CountDownLatch(dockerIds.size());

        for (String dockerId : dockerIds) {
            taskExecutor.submit(() -> {
                try {
                    Container container = DockerClient.inspectContainer(ip, dockerId);
                    if (container == null || "unknown".equals(container.getStatus()) ||
                            (container.getCpu() == 0.0 && container.getMemory() == 0)) {
                        runningCount.getAndDecrement();
                    } else {
                        if (container.getCpu() != null) {
                            usedCpus.getAndAdd(container.getCpu());
                        }
                        if (container.getMemory() != null) {
                            usedMemory.getAndAdd(container.getMemory());
                        }
                    }
                } finally {
                    countDownLatch.countDown();
                }
            });
        }

        try {
            if (countDownLatch.await(3, TimeUnit.SECONDS)) {
                this.runningCount = runningCount.intValue();
                this.usedCpus = usedCpus.doubleValue();
                this.usedMemory = usedMemory.longValue();
                return true;
            }
        } catch (InterruptedException e) {
            log.warn("thread " + Thread.currentThread().getName() + " is interrupted");
        }
        log.warn("host {} sync node usage failed, cause: time out", ip);
        return false;
    }
}
